package com.mjf.spark.day06

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{AccumulatorV2}
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.mutable

/**
 * 累加器
 *    自定义累加器，统计出RDD中，所有以"H"开头的单词以及出现次数(word,count)
 */
object Spark07_Accumulator {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark07_Accumulator")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[String] = sc.makeRDD(List("Hello", "HaHa", "Hello", "HaHa", "Hello", "Spark", "Spark"))

    // 使用自定义累加器-------减少了shuffle过程
    // 创建自定义累加器
    val myAcc: MyAccumulator = new MyAccumulator()

    // 注册累加器
    sc.register(myAcc)

    // 使用累加器
    rdd.foreach(myAcc.add(_))

    // 输出累加器的结果
    println(myAcc.value)

    // 关闭连接
    sc. stop()

  }
}

// 定义一个类，继承AccumulatorV2
// 泛型累加器输入和输出数据的类型
// 闭包检测的时候，先执行copy，再执行reset，再执行isZero
class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Int]]{

  // 定义一个集合，记录单词以及出现次数
  var map = mutable.Map[String, Int]()

  // 是否为初始状态
  override def isZero: Boolean = map.isEmpty

  // 拷贝
  override def copy(): AccumulatorV2[String, mutable.Map[String, Int]] = {
    val newAcc: MyAccumulator = new MyAccumulator()
    newAcc.map = this.map
    newAcc
  }

  // 重置
  override def reset(): Unit = map.clear()

  // 向累加器中添加元素
  override def add(elem: String): Unit = {
    if(elem.startsWith("H")) {
      // 向可变集合添加或者更新元素
      map(elem) = map.getOrElse(elem, 0) + 1
    }
  }

  // 合并
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = {
    // 当前Executor的map
    val map1 = map
    // 另一个Executor的map
    val map2 = other.value

    map = map1.foldLeft(map2) {
      // mm表示map2，kv表示map1中的每个元素
      (mm, kv) => {
        // 指定合并规则
        val k: String = kv._1
        val v: Int = kv._2
        // 根据map1中的元素的key,到map2中获取value
        mm(k) = mm.getOrElse(k, 0) + v // map2中没有的key会进行添加操作
        mm // 将map2返回
      }
    }
  }

  // 获取累加器的值
  override def value: mutable.Map[String, Int] = map
}